package io.kiki.sba.registry.server.metadata.resource;

import io.kiki.sba.registry.common.model.GenericResponse;
import io.kiki.sba.registry.common.model.Response;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.common.model.slot.SlotsStatusResponse;
import io.kiki.sba.registry.server.metadata.MetadataLeaderService;
import io.kiki.sba.registry.server.metadata.lease.dataStoreNode.DataStoreNodeManager;
import io.kiki.sba.registry.server.metadata.multi.cluster.MultiClusterSlotsSyncer;
import io.kiki.sba.registry.server.metadata.multi.cluster.MultiClusterSlotsSyncerImpl.RemoteClusterSlotState;
import io.kiki.sba.registry.server.metadata.resource.filter.LeaderAwareRestController;
import io.kiki.sba.registry.server.metadata.slot.SlotManager;
import io.kiki.sba.registry.server.metadata.slot.arrange.ScheduledSlotArranger;
import io.kiki.sba.registry.server.metadata.slot.status.SlotsStatusService;
import io.kiki.sba.registry.server.metadata.slot.tasks.ReBalanceTaskImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import javax.ws.rs.GET;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;
import java.util.Map;


@Path("openapi/v1/slot/table")
public class SlotsResource {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private SlotManager slotManager;

    @Autowired
    private DataStoreNodeManager dataStoreNodeManager;

    @Autowired
    private ScheduledSlotArranger slotArranger;

    @Autowired
    private MetadataLeaderService metadataLeaderService;

    @Autowired
    private SlotsStatusService SlotsStatusService;

    @Resource
    private MultiClusterSlotsSyncer multiClusterSlotsSyncer;

    public SlotsResource() {
    }

    public SlotsResource(SlotManager slotManager, DataStoreNodeManager dataStoreNodeManager, ScheduledSlotArranger slotArranger, MetadataLeaderService metadataLeaderService, SlotsStatusService SlotsStatusService) {
        this.slotManager = slotManager;
        this.dataStoreNodeManager = dataStoreNodeManager;
        this.slotArranger = slotArranger;
        this.metadataLeaderService = metadataLeaderService;
        this.SlotsStatusService = SlotsStatusService;
    }

    @PUT
    @Path("force/refresh")
    @Produces(MediaType.APPLICATION_JSON)
    @LeaderAwareRestController
    public GenericResponse<Slots> forceRefreshSlots() {
        logger.info("[forceRefreshSlots] begin");
        if (metadataLeaderService.amILeader()) {
            if (slotArranger.tryLock()) {
                try {
                    ReBalanceTaskImpl task = new ReBalanceTaskImpl(slotManager, dataStoreNodeManager.getDataStoreNodeInfo().getList());
                    task.run();
                    logger.info("[forceRefreshSlots] end with succeed");
                    return new GenericResponse<Slots>().fillSucceed(slotManager.getSlots());
                } finally {
                    slotArranger.unlock();
                }
            } else {
                logger.info("[forceRefreshSlots] end, fail to get the lock");
                return new GenericResponse<Slots>().fillFailed("scheduled slot arrangement is running");
            }
        } else {
            logger.info("[forceRefreshSlots] end, not meta-server leader");
            return new GenericResponse<Slots>().fillFailed("not the meta-server leader");
        }
    }

    @PUT
    @Path("/reconcile/stop")
    @Produces(MediaType.APPLICATION_JSON)
    public Response stopSlotsReconcile() {
        logger.info("[stopSlotsReconcile] begin");
        try {
            slotArranger.suspend();
            logger.info("[stopSlotsReconcile] end with succeed");
            return GenericResponse.buildSuccessResponse("succeed");
        } catch (Throwable throwable) {
            logger.error("[stopSlotsReconcile] end", throwable);
            return GenericResponse.buildFailedResponse(throwable.getMessage());
        }
    }

    @PUT
    @Path("/reconcile/start")
    @Produces(MediaType.APPLICATION_JSON)
    public Response startSlotsReconcile() {
        logger.info("[startSlotsReconcile] begin");
        try {
            slotArranger.resume();
            logger.info("[startSlotsReconcile] end with succeed");
            return GenericResponse.buildSuccessResponse("succeed");
        } catch (Throwable throwable) {
            logger.error("[startSlotsReconcile] end", throwable);
            return GenericResponse.buildFailedResponse(throwable.getMessage());
        }
    }

    @GET
    @Path("/reconcile/status")
    @Produces(MediaType.APPLICATION_JSON)
    @LeaderAwareRestController
    public Response getReconcileStatus() {
        logger.info("[getReconcileStatus] begin");
        try {
            boolean result = slotArranger.getLifecycleState().isStarted() && !slotArranger.isSuspended();
            logger.info("[getReconcileStatus] end with succeed");
            if (result) {
                return GenericResponse.buildSuccessResponse("running");
            } else {
                return GenericResponse.buildSuccessResponse("stopped");
            }
        } catch (Throwable throwable) {
            logger.error("[getReconcileStatus] end", throwable);
            return GenericResponse.buildFailedResponse(throwable.getMessage());
        }
    }

    @GET
    @Path("/data/slot/status")
    @Produces(MediaType.APPLICATION_JSON)
    @LeaderAwareRestController
    public GenericResponse<Object> getDataSlotStatuses() {
        logger.info("[getDataSlotStatuses] begin");
        try {
            return new GenericResponse<>().fillSucceed(dataStoreNodeManager.getDataStoreNodeStatsList());
        } catch (Throwable throwable) {
            return new GenericResponse<>().fillFailed(throwable.getMessage());
        } finally {
            logger.info("[getDataSlotStatuses] end");
        }
    }

    @GET
    @Path("/status")
    @Produces(MediaType.APPLICATION_JSON)
    @LeaderAwareRestController
    public GenericResponse<Object> getSlotsStatus() {
        logger.info("[getSlotsStatus] begin");
        try {
            SlotsStatusResponse SlotsStatus = SlotsStatusService.getSlotsStatus();
            return new GenericResponse<>().fillSucceed(SlotsStatus);
        } catch (Throwable th) {
            logger.error("[getSlotsStatus]", th);
            return new GenericResponse<>().fillFailed(th.getMessage());
        } finally {
            logger.info("[getSlotsStatus] end");
        }
    }

    @GET
    @Path("/multi/Slots")
    @Produces(MediaType.APPLICATION_JSON)
    @LeaderAwareRestController
    public GenericResponse<Object> getMultiSlots() {
        logger.info("[getSyncSlotsStatus] begin");
        try {
            Map<String, RemoteClusterSlotState> multiClusterSlots = multiClusterSlotsSyncer.getDataCenterToRemoteClusterSlotStateMap();
            return new GenericResponse<>().fillSucceed(multiClusterSlots);
        } catch (Throwable th) {
            logger.error("[getSyncSlotsStatus]", th);
            return new GenericResponse<>().fillFailed(th.getMessage());
        } finally {
            logger.info("[getSyncSlotsStatus] end");
        }
    }
}
